package AggregateFun;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;

public class ProvinceMAggFun implements AggregateFunction<Tuple4<String,String,Double,Double>, Tuple2<Long,Double>, Tuple2<Long,Double>> {
    //初始化累加器为0
    @Override
    public Tuple2<Long,Double> createAccumulator() {
        return new Tuple2<>(0L,0D); //D表示Double,L表示long
    }

    //把price往累加器上累加
    @Override
    public Tuple2<Long,Double> add(Tuple4<String,String,Double,Double> value, Tuple2<Long,Double> accumulator) {
        double v1=value.f3- value.f2+accumulator.f1;
        long v2=1+accumulator.f0;
        return new Tuple2<>(v2,v1);
    }

    //获取累加结果
    @Override
    public Tuple2<Long,Double> getResult(Tuple2<Long,Double> accumulator) {
        return accumulator;
    }

    //各个subTask的结果合并
    @Override
    public Tuple2<Long,Double> merge(Tuple2<Long,Double> a, Tuple2<Long,Double> b) {
        double v1=a.f1+b.f1;
        long v2 = a.f0+b.f0;
        return new Tuple2<>(v2,v1);
    }
}